package com.cennavi.funcs;

import java.util.HashMap;

import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.broadcast.Broadcast;

import scala.Tuple2;

/** * @author  作者 :allen
    * @date 创建时间：2017年9月29日 上午10:34:22 
    * @version 1.0 
    */

public class RticFlow2RoadPairFunction implements PairFunction<Tuple2<String, String>, String, String> {
	private static final long serialVersionUID = 1L;

	private HashMap<String,String> rtic_roadMap;
	
	public RticFlow2RoadPairFunction(Broadcast<HashMap<String,String>>rtic_roadBroadMap){
		this.rtic_roadMap = rtic_roadBroadMap.value();
	}
	public Tuple2<String, String> call(Tuple2<String, String> arg0) throws Exception {
		if(null != arg0){
			String key_rtic = arg0._2.split(",")[5];
			String key_road = "null";
			if(this.rtic_roadMap.containsKey(key_rtic)){
				key_road = this.rtic_roadMap.get(key_rtic);
			}		
			Tuple2<String, String> returnData = new Tuple2<String, String>(key_road, arg0._2);
			return returnData;
		}else{
			return new Tuple2<String, String>("null", "null");
		}
		
	}

}
